package com.bw.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TableProcess;
import com.bw.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

//1.主流  2.cdc流    3. 输出
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {


//

    private Connection connection;

    //1.创建属性
    private MapStateDescriptor<String, TableProcess> mapState;
    //2.通过构造器给属性赋值
    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapState) {
        this.mapState = mapState;
    }
//
//
//    /*
//    * habse  测试
//    *
//    * */
//    @Override
    public void open(Configuration parameters) throws Exception {
        //phoenix不需要用户密码
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }
//
//



    //主流  放的是数据
    //获取map变量
//    {"database":"gmall","sinkTable":"dim_user_info","table":"user_info","type":"bootstrap-insert","ts":1592788053,
//    "data":{"id":486,"login_name":"d9toe2wto","nick_name":"馨馨","passwd":null,"name":"薛馨","phone_num":"13145774117","email":"d9toe2wto@yahoo.com.cn","head_img":null,"user_level":"1","birthday":"1970-09-14 16:39:30","gender":null,"create_time":"2020-07-14 16:39:30","operate_time":null,"status":null}}
//     {"database":"gmall","table":"user_info","type":"bootstrap-insert","ts":1592788053,"data":{"id":487,"login_name":"uxvp09p","nick_name":"全国","passwd":null,"name":"姜全国","phone_num":"13189875828","email":"uxvp09p@yeah.net","head_img":null,"user_level":"2","birthday":"1989-06-14 16:39:30","gender":"M","create_time":"2020-07-14 16:39:30","operate_time":null,"status":null}}

    //处理主流  包含cdc数据
    //1.获取内存数据和主流进行匹配  过滤出维度数据 放行
    @Override
    public void processElement(JSONObject value, ReadOnlyContext context, Collector<JSONObject> out) throws Exception {



//        //1.获取广播的配置数据
        ReadOnlyBroadcastState<String, TableProcess> bs = context.getBroadcastState(mapState);
        String table = value.getString("table");
//        System.out.println("主流中数据:"+table);
//
//
        TableProcess tableProcess = bs.get(table);
//        System.out.println("主流中数据:"+tableProcess);
//
//        //2.过滤字段  把  不包含的字段删除掉
//        //如果表存储是要主流里面的数据存入到hbase
        if(tableProcess!=null){
            System.out.println("处理之前数据:"+value);
            filterCoulums(value.getJSONObject("data"),tableProcess.getSink_columns());
            System.out.println("处理之后数据:"+value);//{id,name,sinkTable}
            value.put("sinkTable",tableProcess.getSink_table()); //
            out.collect(value);//输出   变名 dim_user_info    name,id
            //
        }else{
            System.out.println("找不到对应的可以:"+table);
        }
//        //3.补充SinkTable的输出
////  {"database":"gmall","table":"base_trademark","type":"insert",
////      "ts":1582301670,"xid":165632,"commit":true,
////      "data":{"id":12,"tm_name":"bw","logo_url":"/static/bw.jpg"}
////  }


    }
    //广播  放的的维度表
    //sBroadcast     Map变量  把你的所有的信息存里面


    /*
    * flinkcdc :  Value:
    *
    * maxwell  产生数据格式   看官网  看文档
    * {"database":"gmall","xid":59965,"data":{"delivery_address":"第11大街第20号楼2单元746门","order_comment":"描述259825","original_total_amount":3973.00,"order_status":"1002","consignee_tel":"13377004018","trade_body":"华为智慧屏V55i-J 55英寸 HEGE-550B 4K全面屏智能电视机 多方视频通话 AI升降摄像头 银钻灰 京品家电等3件商品","id":8056,"operate_time":"2025-06-19 10:49:36","consignee":"尉迟凤","create_time":"2025-06-19 10:49:33","expire_time":"2025-06-19 11:04:33","coupon_reduce_amount":0.00,"out_trade_no":"775854171411261","total_amount":3985.00,"user_id":1422,"img_url":"http://img.gmall.com/334841.jpg","province_id":5,"feight_fee":12.00,"activity_reduce_amount":0.00},"old":{"order_status":"1001"},"xoffset":5940,"type":"update","table":"order_info","ts":1592794177}
     * flinkcdc 产数据格式    看官网  看文档
     {"before":null,
     "after":{
     * "source_table":"sku_info",
     * "sink_table":"dim_sku_info","sink_columns":"id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time","sink_pk":"id","sink_extend":" SALT_BUCKETS = 4"},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1750301620636,"snapshot":"false","db":"gmall_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1750301620636,"transaction":null}

     *
    * */
    //处理cdc  把cdc的数据加载内存中给主流共享
    //1.把cdc 里面的数据创建成维度表
    //2. 把cdc里面的数据存入内存中
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
//
//        //1.解析数据
        JSONObject jsonObject = JSON.parseObject(value);
        //把json数据映射对象里面
        //对象里面的字段名称和json里面一直才能成功
        TableProcess tp = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
//        System.out.println(tp);
//
//
//        //2.校验并且建表判断phoenix--->habse 是否存在表 (类似于jdbc)
        //1.获取连接  jdbc    2.编写sql    3.执行sql   4. ps
        checkTable(tp.getSink_table(),tp.getSink_columns(),tp.getSink_pk(),tp.getSink_extend());
//        //--->
//        //3.写入状态  广播出去(自己做)
        //
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapState);
        broadcastState.put(tp.getSource_table(),tp);


//        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapState);
//        broadcastState.put(tp.getSource_table(),tp);

    }
    private void filterCoulums(JSONObject data, String sink_columns) {
        String[] split = sink_columns.split(",");
        //把数组转换成列表
        List<String> list = Arrays.asList(split);
//        "data":{"id":12,"tm_name":"bw","logo_url":"/static/bw.jpg"}
//         id,tm_name
         //对数据进行遍历  把json当中map
        Set<Map.Entry<String, Object>> entries = data.entrySet();



         //迭代器
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            //这个字符就是包含的意思
            //如果把字符串转换成list    那么就判断是否包含元素
            if(!list.contains(next.getKey())){
                //不包含的话把key-value删除掉
                iterator.remove();
            }
        }

    }
//
//
//
//    /*
//     * create  table  if  not  exists   tablename(
//     *id  varchar  primary  key,
//     * bb  varchar,
//     * cc  varchar
//     * );
//     *
//     * */
    private void checkTable(String sink_table, String sink_columns, String sink_pk, String sink_extend)  {


        /*
        * 1.获取连接（连接池）
        * 2.获取ps
        * 3.发送sql
        * 4.执行sql
        *
        *
        *
        *
        * */


        PreparedStatement ps=null;



        try{
            //处理字段
            if(sink_pk==null||"".equals(sink_pk)){
                sink_pk="id";
            }
            if (sink_extend==null){
                sink_extend="";
            }
            //拼接sql

            /*

            *
            *c
            *
            *
            * */
            StringBuilder createTable = new StringBuilder("CREATE  TABLE  IF  NOT EXISTS    ")
                    .append(GmallConfig.HBASE_SCHEMA)   //not  exists    schema "GMALL_REALTIME"
                    .append(".")
                    .append(sink_table)
                    .append("(");
//            id,activity_name,activity_type,activity_desc,start_time,end_time,create_time
//            id,activity_name,activity_type,activity_desc,start_time,end_time,create_time
            String[] split = sink_columns.split(",");

            for (int i = 0; i < split.length; i++) {
                //取出字段

                //是否为主键
                if(sink_pk.equals(split[i])){
                    createTable.append(split[i]).append("  varchar PRIMARY KEY");
                }else{
                    createTable.append(split[i]).append("  varchar");
                }
                //判断是否是最后一个字段
                if(i<split.length-1){
                    createTable.append(",");
                }

            }
            createTable.append(")").append(sink_extend);

            System.out.println("创建表语句为:"+createTable);

            String sql = createTable.toString();



            //mysql  clickhouse  hbase   connection(数据源)

            ps = connection.prepareStatement(sql);
            ps.execute();


            //执行sql //正常数据
        }catch (Exception e){
            //把编译时异常  变成运行时异常（程序不会中断）
            new RuntimeException("建表失败:"+sink_table);
        }finally {
            if(ps!=null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }



    }


}

